package com.erainm.consumerInterceptors;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;

/**
 * Created by erainm on 2021/9/11 5:57 下午.
 *
 * @Description TODO:
 **/
public class ThreeInterceptor implements ConsumerInterceptor<String,String> {
    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> records) {
        // poll 方法返回结果之前最后要调用的方法
        System.out.println("Three -- 开始");
        // 不做处理，直接返回
        return records;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 消费者提交偏移量的时候，经过该方法
        System.out.println("Three -- 结束");

    }

    @Override
    public void close() {
        // 用于关闭拦截器用到的资源，如打开文件，连接的数据库等
    }

    @Override
    public void configure(Map<String, ?> configs) {
        // 用于获取消费者的设置参数
        configs.forEach((k,v) -> {
            System.out.println(k + "\t" + v);
        });
    }
}
